/**
 * 
 */
package org.pump.client.type;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.mina.core.buffer.IoBuffer;
import org.pump.client.codec.BinlogEvent;
import org.pump.client.codec.EventData;
import org.pump.client.codec.Packet;
import org.pump.client.codec.TableMapEvent;
import org.pump.client.util.TypeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author liudong03
 * 
 */
public class RowValue {
	private static final Logger mylog = LoggerFactory.getLogger(RowValue.class);
	boolean[] column_null;
	List<Object> column_value;
	List<Integer> column_pos;
	int rowlen;
	public static String charset = "gbk";
	
	
	public RowValue() {
		this.column_value = new ArrayList<Object>();
		this.column_pos = new ArrayList<Integer>();
	}

	public static RowValue readBuffer(IoBuffer in, int len, Packet prevpacket,
			boolean[] use_column) {
		//
		if (prevpacket == null) {
			mylog.error("prev packet is null");
			return null;
		}
		if (in.remaining() < len) {
			return null;
		}
		RowValue myrow = new RowValue();

		// 获取前一个packet，这个packet我们假定会是TableMapEvent,如果不是，我们会报错
		EventData prev_event_data = ((BinlogEvent) prevpacket.getBody())
				.getMydata();
		if (!(prev_event_data instanceof TableMapEvent)) {
			mylog.error("the priveous packet is not TableMapEvent");
			return null;
		}
		TableMapEvent mapevent = (TableMapEvent) prev_event_data;

		// 计算列数
		int column_count = use_column.length;
		// 计算使用列的列数
		int used_column_count = 0;
		for (boolean curcol : use_column) {
			if (curcol)
				used_column_count++;
		}
		// 为column_null分配内存
		myrow.column_null = new boolean[used_column_count];
		// 计算表示 column_null需要读取的字节数目
		int null_value_size = (used_column_count + 7) >> 3;
		// 分配读取column_null空间
		byte[] temp_null_value = new byte[null_value_size];
		in.get(temp_null_value);
		myrow.rowlen += null_value_size;
		int counter = 0;
		// 解析temp_null_value内容
		for (int i = 0; i < column_count; i++) {
			if (use_column[i]) {
				myrow.column_null[counter] = ((temp_null_value[counter >> 3] & (0x01 << (counter % 8))) == 0) ? false
						: true;
				counter++;
			}
		}

		// 读取内容
		counter = 0;
		for (int i = 0; i < column_count; i++) {
			// 当此列被使用
			if (use_column[i]) {
				// 而且此列非空
				if (!myrow.column_null[counter]) {
					Short coltype_obj = mapevent.getColumn_type().get(i);
					if (coltype_obj == null) {
						mylog.error("coltype_obj is null");
						return null;
					}
					short coltype = coltype_obj.shortValue();

					byte[] colmeta = mapevent.getMetadata().get(i);

					if (!RowValue.getData(in, coltype, colmeta, myrow)) {
						mylog.error("read data value fail");
						return null;
					}

					Integer pos = new Integer(i);
					myrow.column_pos.add(pos);
				}
				counter++;
			}
		}

		return myrow;
	}

	public static boolean getData(IoBuffer in, long type, byte[] meta,
			RowValue myrow) {
		long length = 0;
		if (type == MysqlType.MYSQL_TYPE_STRING) {
			if (meta.length != 2) {
				mylog.error(type + "");
				return false;
			}
			long byte0 = meta[0] & 0xff;
			long byte1 = meta[1] & 0xff;
			if (byte0 != 0) {
				if ((byte0 & 0x30) != 0x30) {
					length = byte1 | (((byte0 & 0x30) ^ 0x30) << 4);
					type = byte0 | 0x30;
				} else {
					switch ((int) byte0) {
					case MysqlType.MYSQL_TYPE_SET:
					case MysqlType.MYSQL_TYPE_ENUM:
					case MysqlType.MYSQL_TYPE_STRING:
						type = byte0;
						length = byte1;
						break;
					default: {
						mylog.error("unsupported mysql type");
						return false;
					}
					}
				}
			} else {
				length = byte1;
			}
		}

		switch ((int) type) {
		case MysqlType.MYSQL_TYPE_LONG: {
			int data = in.getInt();
			Integer mydata = new Integer(data);

			myrow.column_value.add(mydata);
			myrow.rowlen += 4;

			return true;
		}
		case MysqlType.MYSQL_TYPE_TINY: {
			short data = in.get();
			Short mydata = new Short(data);

			myrow.column_value.add(mydata);
			myrow.rowlen += 1;
			return true;
		}
		case MysqlType.MYSQL_TYPE_SHORT: {
			short data = in.getShort();
			Short mydata = new Short(data);

			myrow.column_value.add(mydata);
			myrow.rowlen += 2;
			return true;
		}
		case MysqlType.MYSQL_TYPE_INT24: {
			int data = in.getMediumInt();
			Integer mydata = new Integer(data);

			myrow.column_value.add(mydata);
			myrow.rowlen += 3;

			return true;
		}
		case MysqlType.MYSQL_TYPE_LONGLONG: {
			long data = in.getLong();
			Long mydata = new Long(data);

			myrow.column_value.add(mydata);
			myrow.rowlen += 8;

			return true;
		}
		case MysqlType.MYSQL_TYPE_NEWDECIMAL: {
			if (meta.length != 2) {
				mylog.info("new decimal meta length is not 2:" + meta.length);
				return false;
			}
			int percision = meta[0] & 0xff;
			int decimals = meta[1] & 0xff;
			int sz = TypeUtil.calc_decimal_bytes(percision, decimals);

			byte[] data = new byte[sz];
			in.get(data);
			MysqlDecimal dt = new MysqlDecimal(data,percision,decimals);
			myrow.column_value.add(dt);
			myrow.rowlen += sz;

			return true;
		}
		case MysqlType.MYSQL_TYPE_FLOAT: {
			float data = in.getFloat();
			Float mydata = new Float(data);

			myrow.column_value.add(mydata);
			myrow.rowlen += 4;
			return true;
		}
		case MysqlType.MYSQL_TYPE_DOUBLE: {
			double data = in.getDouble();
			Double mydata = new Double(data);

			myrow.column_value.add(mydata);
			myrow.rowlen += 8;
			return true;
		}
		case MysqlType.MYSQL_TYPE_BIT: {
			long byte0 = meta[0] & 0xff;
			long byte1 = meta[1] & 0xff;
			long nbits = (byte1 * 8) + (byte0 & 0xFF);
			length = (nbits + 7) / 8;

			byte[] data = new byte[(int) length];
			in.get(data);
			MysqlBit dt = new MysqlBit(data);
			myrow.column_value.add(dt);
			myrow.rowlen += length;
			// mylog.info("developing");
			return true;
		}
		case MysqlType.MYSQL_TYPE_TIMESTAMP: {
			long data = in.getUnsignedInt();
			MysqlTimestamp st = new MysqlTimestamp(data);

			myrow.column_value.add(st);
			myrow.rowlen += 4;
			return true;
		}
		case MysqlType.MYSQL_TYPE_DATETIME: {
			long data = in.getLong();

			MysqlDateTime dt = new MysqlDateTime(data);
			myrow.column_value.add(dt);
			myrow.rowlen += 8;

			return true;
		}
		case MysqlType.MYSQL_TYPE_TIME: {
			int data = in.getUnsignedMediumInt();

			MysqlTime dt = new MysqlTime(data);
			myrow.column_value.add(dt);
			myrow.rowlen += 3;

			return true;
		}
		case MysqlType.MYSQL_TYPE_DATE: {
			int data = in.getUnsignedMediumInt();

			MysqlDate dt = new MysqlDate(data);
			myrow.column_value.add(dt);
			myrow.rowlen += 3;

			return true;
		}
		case MysqlType.MYSQL_TYPE_YEAR: {
			short data = in.getUnsigned();

			MysqlYear dt = new MysqlYear(data);
			myrow.column_value.add(dt);
			myrow.rowlen += 1;

			return true;
		}
		case MysqlType.MYSQL_TYPE_ENUM: {
			int data = 0;
			switch ((int) length) {
			case 1: {
				data = in.getUnsigned();
				MysqlEnum dt = new MysqlEnum(data, (int) length);
				myrow.column_value.add(dt);
				myrow.rowlen += 1;
				return true;
			}
			case 2: {
				data = in.getUnsignedShort();
				MysqlEnum dt = new MysqlEnum(data, (int) length);
				myrow.column_value.add(dt);
				myrow.rowlen += 2;
				return true;
			}
			default: {
				mylog.error("unsupported enum size");
				return false;
			}
			}
		}
		case MysqlType.MYSQL_TYPE_SET: {
			byte[] data = new byte[(int) length];
			in.get(data);
			MysqlSet dt = new MysqlSet(data, (int) length);

			myrow.column_value.add(dt);
			myrow.rowlen += length;
			return true;

		}
		case MysqlType.MYSQL_TYPE_BLOB: {
			if (meta.length != 1) {
				mylog.error("the meta size is not 1");
				return false;
			}
			int blob_size = meta[0] & 0xff;
			switch (blob_size) {
			case 1: {
				length = in.getUnsigned();
				myrow.rowlen++;
				
				byte[] data = new byte[(int) length];
				in.get(data);
				MysqlBlob mydata = new MysqlBlob(data);
				myrow.column_value.add(mydata);
				myrow.rowlen += length;

				return true;
			}
			case 2: {
				length = in.getUnsignedShort();
				myrow.rowlen += 2;

				byte[] data = new byte[(int) length];
				in.get(data);
				MysqlBlob mydata = new MysqlBlob(data);
				myrow.column_value.add(mydata);
				myrow.rowlen += length;
				return true;
			}
			case 3: {
				length = in.getUnsignedMediumInt();
				myrow.rowlen += 3;

				byte[] data = new byte[(int) length];
				in.get(data);
				MysqlBlob mydata = new MysqlBlob(data);
				myrow.column_value.add(mydata);
				myrow.rowlen += length;
				return true;
			}
			case 4: {
				length = in.getUnsignedInt();
				myrow.rowlen += 4;

				byte[] data = new byte[(int) length];
				in.get(data);
				MysqlBlob mydata = new MysqlBlob(data);
				myrow.column_value.add(mydata);
				myrow.rowlen += length;
				return true;
			}
			default: {
				mylog.info("unsupported blob size:" + length);
				return false;
			}
			}
		}
		case MysqlType.MYSQL_TYPE_VARCHAR:
		case MysqlType.MYSQL_TYPE_VAR_STRING: {
			length = TypeUtil.calc_meta(meta);
			if (length < 256) {
				short sz = in.getUnsigned();
				myrow.rowlen++;
				Varstr mydata = Varstr.readBuffer(in, sz, RowValue.charset);
				myrow.column_value.add(mydata);
				myrow.rowlen += mydata.bytelen;
			} else {
				int sz = in.getUnsignedShort();
				myrow.rowlen += 2;
				Varstr mydata = Varstr.readBuffer(in, sz, RowValue.charset);
				if(mydata == null)
				{
					mylog.error("read varstr fail");
				}
				myrow.column_value.add(mydata);
				myrow.rowlen += mydata.bytelen;
			}
			return true;
		}
		case MysqlType.MYSQL_TYPE_STRING: {
			if (length < 256) {
				short sz = in.getUnsigned();
				myrow.rowlen++;
				Varstr mydata = Varstr.readBuffer(in, sz, RowValue.charset);
				myrow.column_value.add(mydata);
				myrow.rowlen += mydata.bytelen;
			} else {
				int sz = in.getUnsignedShort();
				myrow.rowlen += 2;
				Varstr mydata = Varstr.readBuffer(in, sz, RowValue.charset);
				myrow.column_value.add(mydata);
				myrow.rowlen += mydata.bytelen;
			}
			return true;
		}
		default: {
			mylog.info("unsupported type:[" + type + "]");
			return false;
		}
		}

	}

	/* (non-Javadoc)
	 * @see java.lang.Object#toString()
	 */
	@Override
	public String toString() {
		return "RowValue [column_null=" + Arrays.toString(column_null)
				+ ", column_value=" + column_value + ", column_pos="
				+ column_pos + ", rowlen=" + rowlen + "]";
	}

	/**
	 * @return the rowlen
	 */
	public int getRowlen() {
		return rowlen;
	}

	/**
	 * @return the column_null
	 */
	public boolean[] getColumn_null() {
		return column_null;
	}

	/**
	 * @return the column_value
	 */
	public List<Object> getColumn_value() {
		return column_value;
	}

	/**
	 * @return the column_pos
	 */
	public List<Integer> getColumn_pos() {
		return column_pos;
	}
	
	
}
